package day02;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

public class Example2 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //env.getConfig().setAutoWatermarkInterval(60 * 1000L);

        env
                .socketTextStream("hadoop102",7777)
                .map(new MapFunction<String, Tuple2<String,Long>>() {
                         @Override
                         public Tuple2<String, Long> map(String value) throws Exception {
                             String[] arr = value.split(" ");
                             return Tuple2.of(arr[0],Long.parseLong(arr[1]) * 1000L);
                         }
                     })
                .keyBy(r -> r.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> value, Context ctx, Collector<String> out) throws Exception {
                        long currentTs = ctx.timerService().currentProcessingTime();
                        out.collect("到达的数据是：" + value + "，当前处理时间是：" + new Timestamp(currentTs));
                        // 注册一个10秒钟之后的定时器
                        long tenSecLater = currentTs + 10 * 1000L;
                        ctx.timerService().registerProcessingTimeTimer(tenSecLater);
                        out.collect("注册了一个机器时间是：" + new Timestamp(tenSecLater) + "的定时器");
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("定时器触发了！触发时间是：" + new Timestamp(timestamp) + ctx.timerService().currentWatermark());
                    }
                })
                .print();

        env.execute();

    }
}
